package com.gagakuai.domain.jude.expression;

import com.gagakuai.domain.conditions.Condition;
import com.gagakuai.domain.facts.Fact;
import com.gagakuai.domain.facts.Facts;
import com.gagakuai.domain.index.calculate.IndexCalculateConfig;
import com.gagakuai.domain.index.RuleIndex;
import com.gagakuai.domain.index.datareader.DataReader;
import com.gagakuai.domain.index.datasource.AbstractIndexDataSource;
import com.gagakuai.domain.jude.RuleJudeExecutor;
import com.gagakuai.domain.rules.Rule;
import com.gagakuai.infrastructure.utils.FlinkUtils;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @description: 基类
 * @author: houhong
 * @create: 2024-09-07 14:24
 **/
public abstract class AbstractRuleJudeExecutor<T extends Rule> implements RuleJudeExecutor<T>, SmartInitializingSingleton {

    protected final Logger LOGGER = LoggerFactory.getLogger(this.getClass());

    @Autowired
    List<DataReader> dataReaders;
    Map<Integer, DataReader> dataReaderMap = new HashMap<>();

    /*
     * 从数据源读取数据和传入的facts 一起，然后一起进行指标计算。
     */
    public Facts indexDataReadAndCalculate(T rule, Facts facts) {

        Object partyId = facts.get("partyId");

        List<Condition> conditions = rule.getConditions();

        List<RuleIndex> ruleIndexes = conditions.stream()
                .map(Condition::getRuleIndex)
                .collect(Collectors.toList());

        Facts indexFacts = new Facts();
        for (RuleIndex ruleIndex : ruleIndexes) {

            AbstractIndexDataSource dataSource = ruleIndex.getDataSource();
            dataSource.setPartyId(partyId);
            //获取数据
            Facts factFromDataSource = getFactsFromDataSource(dataSource);
            facts.add(factFromDataSource);
            //将数据计算成指标，存入facts
            Facts index = calculateFacts(ruleIndex.getCalculateConfig(), factFromDataSource);
            indexFacts.add(index);
        }
        return indexFacts;
    }


    /*
     * 从数据源获取数据
     */
    private Facts getFactsFromDataSource(AbstractIndexDataSource dataSource) {

        Integer sourceType = dataSource.getType();

        DataReader dataReader = dataReaderMap.get(sourceType);
        if (dataReader == null) {
            LOGGER.error("数据源类型:{}, 未找到对应对的 数据读取器", sourceType);
            return null;
        }
        return dataReader.readData(dataSource);

    }

    /*
     * 这里针对指标的计算，
     *
     *   第一版本：
     *        使用groovy 作为指标在线计算脚本,根据指标配置的数据源信息
     *        获取到相关数据后，然后根据配置界面 配置的指标计算脚本，使用groovy进行指标计算。
     *        因为指标有需要计算的，有不需要计算的，针对需要进行计算的指标，会存入到redis中。
     *        由于数据都是t+1，所以缓存时间设置为24 * 60 * 60，为提高性能，使用定时任务，预先计算放入到redis/hbase中
     *
     *         针对不需要计算的数据:(存入到redis 数据量太大，比如 员工的体检信息数据 整体有几百万，但是每个人只有300多条）
     *         实时读取出来 将指标作为 key- 读取到的值作为 value ： 放入到Facts中 让后面进行规则执行器执行
     *
     *
     *
     *     第二版本：
     *      1:数据 通过离线 写入到业务表中，flink 通过cdc 监听相关的表，然后根据后台配置的
     *      指标规则，实时计算指标的值，然后 根据指标配置去redis/hbase 中获取指标
     *      2: 如果没有读取到指标，那么就根据配置去数据库里去读取数据 然后使用flink/spark进行在线批处理计算
     *      再将指标写入redis/hbase中
     *      3: 将 指标放入到facts中 放到下游进行，放入到规则判断器中去执行
     *
     * // todo 这里涉及到公司敏感，暂不公开
     *    当前关注于规则的配置和执行
     */
    private Facts calculateFacts(IndexCalculateConfig calculateConfig, Facts facts) {

        Facts indexFacts = new Facts();
        /*
         *
         *  这里目前版本
         */
        Set<Fact<?>> factsSet = facts.getFacts();
        //
        DataSet<Fact<?>> dataStreamSource = FlinkUtils.BATCH_ENV.fromCollection(factsSet);


        return indexFacts;

    }

    @Override
    public void afterSingletonsInstantiated() {

        if (CollectionUtils.isEmpty(dataReaders)) return;

        dataReaderMap = dataReaders.stream()
                .collect(Collectors.toMap(DataReader::getType, dataReader -> dataReader));
    }
}
